package com.activeMq.producer.controller;

import com.activeMq.producer.service.ActviceMqSendService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

@Controller
public class ProducerController {

    @Autowired
    private ActviceMqSendService actviceMqSendService;

    private final Map<String, Object> resultMap = new HashMap<String, Object>() {{
        put("success", true);
        put("message", "");
    }};

    private ExecutorService threadPool = null; //线程池
    private AtomicInteger sendCount = new AtomicInteger();//提交记数
    private AtomicInteger successCount = new AtomicInteger();//发送成功记数
    private AtomicInteger failCount = new AtomicInteger();//发送失败记数
    private long beginTime = 0;//开始时间
    private boolean sendDown = false;//false 没有发送 true发送完成
    private List<Future<Map<String, Object>>> taskResult = new ArrayList<>();//任务结果

    @GetMapping("")
    public String Index() {
        return "Index";
    }

    @PostMapping("startSend")
    @ResponseBody
    public Map<String, Object> startSend(int taskNum, int maxNum) {
        sendCount.set(0);
        successCount.set(0);
        failCount.set(0);
        sendDown = false;
        beginTime = System.nanoTime();
        threadPool = Executors.newFixedThreadPool(taskNum);
        new Thread(() -> productionDispatch(maxNum)).start();//线程提交任务
        resultMap.put("message", "开始生产消息到ActiveMq");
        return resultMap;
    }

    @PostMapping("stopSend")
    @ResponseBody
    public Map<String, Object> stopSend() {
        if (threadPool != null)
            threadPool.shutdownNow();
        taskResult.clear();
        resultMap.put("message", "ActiveMq停止生产消息");
        return resultMap;
    }

    @PostMapping("getMsgCount")
    @ResponseBody
    public Map<String, Object> getMsgCount() {
        Map<String, Object> dataMap = new HashMap<>();
        dataMap.put("noIndex", sendCount);
        dataMap.put("successCount", successCount);
        dataMap.put("failCount", failCount);
        dataMap.put("timeCounnt", TimeUnit.NANOSECONDS.toSeconds((System.nanoTime() - beginTime)) + "秒");
        dataMap.put("sendDown", sendDown);
        resultMap.put("message", "发送进度");
        resultMap.put("data", dataMap);
        return resultMap;
    }

    //[线程调用]提交任务
    private void productionDispatch(Integer maxCount) {
        for (int i = 0; i < maxCount; i++) {
            Future<Map<String, Object>> future = threadPool.submit(getCallable());
            taskResult.add(future);
        }
        System.out.println("提交任务耗时:" + TimeUnit.NANOSECONDS.toSeconds((System.nanoTime() - beginTime)) + "毫秒");
        new Thread(this::checkSendResult).start();//检测结果
    }

    //[线程调用]检查任务结果
    private void checkSendResult() {
        boolean flag = true;
        while (flag) {
            Iterator<Future<Map<String, Object>>> iter = taskResult.iterator();
            int i = 0;
            while (iter.hasNext()) {
                if (i % 10 == 0) {
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                Future<Map<String, Object>> f = iter.next();
                if (f.isDone()) {
                    try {
                        if ((boolean) f.get().get("success")) {
                            successCount.addAndGet(1);
                        } else {
                            failCount.addAndGet(1);
                        }
                        iter.remove();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                i++;
            }
            if (taskResult.size() == 0) {
                flag = false;
            }
        }
        sendDown = true;
        threadPool.shutdownNow();
        System.out.println("检测结果完毕!");
    }


    //生成单个消息任务
    private Callable<Map<String, Object>> getCallable() {
        return () -> {
            sendCount.addAndGet(1);
            Map<String, Object> uuidMap = new HashMap<>();
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            uuidMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
            Map<String, Object> map = new HashMap<>();
            map.put("No.", sendCount.get());
            map.put("Msg", uuidMap);
            return actviceMqSendService.sendMessage(map);
        };
    }


    public static void main(String[] args) {
        //System.out.println(TimeUnit.NANOSECONDS.toSeconds(10000000000L));
        System.out.println(101%10);
    }
}
